package ru.unlocker.reports.kafka;

import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import lombok.Getter;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Getter
public class KafkaConfiguration {

  protected static final Logger logger = LoggerFactory.getLogger(KafkaConfiguration.class.getName());

  private String reportsTopicName = "reports";
  private String timersTopicName = "timers";
  private String bootstrapServers = "kafka1:9092";
  private String truststoreLocation = "";
  private String truststorePassword = "";

  public KafkaConfiguration() {
  }

  public KafkaConfiguration(String bootStrap) {
    this.bootstrapServers = bootStrap;
  }

  static Optional<String> extractConfigValue(Map<String, String> env, String key) {
    return Optional.ofNullable(env.get(key)).filter(s -> !s.isBlank());
  }

  /**
   * @return common kafka properties
   */
  private Properties buildCommonProperties(Map<String, String> env) {
    Properties properties = new Properties();

    extractConfigValue(env, "kafka.topic.reports").ifPresent(s -> reportsTopicName = s);
    extractConfigValue(env, "kafka.topic.timers").ifPresent(s -> timersTopicName = s);
    extractConfigValue(env, "kafka.bootstrap.servers").ifPresent(s -> bootstrapServers = s);
    properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

    Optional<String> v = extractConfigValue(env, "kafka.security.protocol");
    v.ifPresent(s -> properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, s));
    v = extractConfigValue(env, "kafka.sasl.mechanism");
    v.ifPresent(s -> properties.put(SaslConfigs.SASL_MECHANISM, s));

    v = extractConfigValue(env, "kafka.sasl.jaas.config");
    v.ifPresent(s -> properties.put(SaslConfigs.SASL_JAAS_CONFIG, s));
    properties.put(SslConfigs.SSL_PROTOCOL_CONFIG, "TLSv1.2");
    properties.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, "TLSv1.2");
    properties.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "HTTPS");

    v = extractConfigValue(env, "kafka.ssl.truststore.location");
    if (v.isPresent()) {
      properties.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, v.get());
      truststoreLocation = v.get();
    }

    v = extractConfigValue(env, "kafka.ssl.truststore.password");
    if (v.isPresent()) {
      properties.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, v.get());
      truststorePassword = v.get();
    }

    v = extractConfigValue(env, "kafka.ssl.keystore.location");
    v.ifPresent(s -> properties.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, s));

    v = extractConfigValue(env, "kafka.ssl.keystore.password");
    v.ifPresent(s -> properties.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, s));

    return properties;
  }


  public Properties getConsumerProperties(String groupId) {
    final var env = System.getenv();
    Properties properties = buildCommonProperties(env);
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    Optional<String> vs = extractConfigValue(env, "kafka.value.deserializer");
    if (vs.isPresent()) {
      properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, vs.get());
    } else {
      properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
          "org.apache.kafka.common.serialization.ByteArrayDeserializer");
    }

    vs = extractConfigValue(env, "kafka.key.deserializer");
    if (vs.isPresent()) {
      properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, vs.get());
    } else {
      properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
          "org.apache.kafka.common.serialization.ByteArrayDeserializer");
    }
    return properties;
  }

  public Properties getProducerProperties(String clientID) {
    final var env = System.getenv();
    Properties properties = buildCommonProperties(env);

    Optional<Long> v = extractConfigValue(env, "kafka.producer.timeout.sec").map(Long::parseLong);
    v.ifPresent(aLong -> properties.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, aLong));

    Optional<String> vs = extractConfigValue(env, "kafka.producer.acks");
    if (vs.isPresent()) {
      properties.put(ProducerConfig.ACKS_CONFIG, vs.get());
    } else {
      properties.put(ProducerConfig.ACKS_CONFIG, "all");
    }

    Optional<Boolean> vb = extractConfigValue(env, "kafka.producer.idempotence").map(Boolean::parseBoolean);
    if (vb.isPresent()) {
      properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, vb.get());
    } else {
      properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false);
    }

    Optional<Integer> vi = extractConfigValue(env, "kafka.producer.retries").map(Integer::parseInt);
    if (vi.isPresent()) {
      properties.put(ProducerConfig.RETRIES_CONFIG, vi.get());
    } else {
      properties.put(ProducerConfig.RETRIES_CONFIG, 0);
    }

    vs = extractConfigValue(env, "kafka.key.serializer");
    if (vs.isPresent()) {
      properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, vs.get());
    } else {
      properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
          "org.apache.kafka.common.serialization.ByteArraySerializer");
    }

    vs = extractConfigValue(env, "kafka.value.serializer");
    if (vs.isPresent()) {
      properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, vs.get());
    } else {
      properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
          "org.apache.kafka.common.serialization.ByteArraySerializer");
    }

    properties.put(ProducerConfig.CLIENT_ID_CONFIG, clientID);

    properties.forEach((k, val) -> logger.info(k + " : " + val));
    return properties;
  }

}
